package com.study.kafkademo01.service.impl;

import com.study.kafkademo01.config.KafkaConfig;
import com.study.kafkademo01.service.KafkaProducerService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.stereotype.Service;

import java.util.Properties;

/**
 * @author weixiaochun
 */
@Service
//@Slf4j
public class KafkaProducerServiceImpl implements KafkaProducerService {

    @Override
    public void sendMessage(String message) {
        //1、加载Kafka配置文件
        Properties properties = KafkaConfig.getKafkaProperties();

        //2、创建生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //3、发送数据
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>(
                    "first","study" + i
                    );
            int finalI = i;
            producer.send(producerRecord, (metadata, exception) -> {
                if(exception == null){
                    System.out.println("metadata：" + metadata.toString());
                    System.out.println(metadata.partition() + "---" + metadata.offset() + "---" + finalI);
                }else{
                    exception.printStackTrace();
                }
            });
        }

        //4、关闭资源
        producer.close();

    }
}
